package com.tt.job.rocketmq;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.tt.job.api.JobContext;
import com.tt.job.common.util.JsonUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class MqProducer {

    private final static Logger logger = LoggerFactory.getLogger(MqProducer.class);
    @Autowired
    MqSettings settings;

    private DefaultMQProducer producer = null;
    private MessageQueueSelector queueSelector = new SelectMessageQueueByHash();
    private boolean started = false;

    /**
     * 发送指定调度任务消息
     *
     * @param jobContext
     * @param delayLevel
     * @return
     * @throws InterruptedException
     * @throws RemotingException
     * @throws MQClientException
     * @throws MQBrokerException
     */
    public boolean produce(JobContext jobContext, int delayLevel) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        if (producer == null) {
            return false;
        }
        //考虑到任务不知道自己的调度方式，只能使用job的信息来进行消息发送
        // 使用需要调度的Job的name来作为tags方便消费者订阅
        Message message = new Message();
        String tag = jobContext.getJobClass().replace(".", "-");
        message.setTopic(settings.getTopic());
        message.setTags(tag);
        message.setKeys(jobContext.getSchedule().getKey());
        message.setBody(JsonUtil.encodeByte(jobContext));
        message.setDelayTimeLevel(delayLevel);
        SendResult result;
        if (StringUtils.isNotEmpty(jobContext.getSchedule().getKey())) {
            result = producer.send(message, queueSelector, jobContext.getSchedule().getKey());
        } else {
            result = producer.send(message);
        }
        logger.debug("mq producer produce message:{}, result:{}", message, result);
        return result.getSendStatus() == SendStatus.SEND_OK;
    }

    public void start() throws Exception {
        synchronized (this) {
            if (!started) {
                producer = new DefaultMQProducer();
                producer.setProducerGroup(settings.getGroup());
                producer.setNamesrvAddr(settings.getNameServer());
                this.producer.start();
                started = true;
                logger.info("mq producer started!");
            }
        }
    }

    public void stop() {
        synchronized (this) {
            if (started) {
                started = false;
                this.producer.shutdown();
                this.producer = null;
                logger.info("mq producer stop!");
            }
        }
    }
}
